package com.example.flinkcourse.lesson1.state;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * 状态管理类
 * 配置状态后端和检查点
 */
public class StateAndCheckpointManager {
    private static final Logger LOG = LoggerFactory.getLogger(StateAndCheckpointManager.class);

    /**
     * 配置状态后端和检查点
     * @param env StreamExecutionEnvironment
     */
    public static void configureStateAndCheckpoint(StreamExecutionEnvironment env) {
        try {
            // 配置检查点
            env.enableCheckpointing(60000); // 每60秒做一次检查点
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
            env.getCheckpointConfig().setCheckpointTimeout(60000);
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
            env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                org.apache.flink.streaming.api.CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            // 配置状态后端
            StateBackend stateBackend = new RocksDBStateBackend("hdfs://localhost:9000/flink/checkpoints", true);
            env.setStateBackend(stateBackend);

            // 配置重启策略
            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3, // 重启次数
                Time.of(10, TimeUnit.SECONDS) // 重启延迟
            ));

            LOG.info("State and checkpoint configuration completed");
        } catch (IOException e) {
            LOG.error("Failed to configure state and checkpoint", e);
            throw new RuntimeException("Failed to configure state and checkpoint", e);
        }
    }
} 